{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Initialize PySpark\n",
    "\n",
    "First, we use the findspark package to initialize PySpark."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Initialize PySpark\n",
    "APP_NAME = \"Debugging Prediction Problems\"\n",
    "\n",
    "# If there is no SparkSession, create the environment\n",
    "try:\n",
    "  sc and spark\n",
    "except NameError as e:\n",
    "  import findspark\n",
    "  findspark.init()\n",
    "  import pyspark\n",
    "  import pyspark.sql\n",
    "\n",
    "  sc = pyspark.SparkContext()\n",
    "  spark = pyspark.sql.SparkSession(sc).builder.appName(APP_NAME).getOrCreate()\n",
    "\n",
    "print(\"PySpark initiated...\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Hello, World!\n",
    "\n",
    "Loading data, mapping it and collecting the records into RAM..."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Load the text file using the SparkContext\n",
    "csv_lines = sc.textFile(\"../data/example.csv\")\n",
    "\n",
    "# Map the data to split the lines into a list\n",
    "data = csv_lines.map(lambda line: line.split(\",\"))\n",
    "\n",
    "# Collect the dataset into local RAM\n",
    "data.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Creating Objects from CSV using `pyspark.RDD.map`\n",
    "\n",
    "Using a function with a map operation to create objects (dicts) as records..."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Turn the CSV lines into objects\n",
    "def csv_to_record(line):\n",
    "  parts = line.split(\",\")\n",
    "  record = {\n",
    "    \"name\": parts[0],\n",
    "    \"company\": parts[1],\n",
    "    \"title\": parts[2]\n",
    "  }\n",
    "  return record\n",
    "\n",
    "# Apply the function to every record\n",
    "records = csv_lines.map(csv_to_record)\n",
    "\n",
    "# Inspect the first item in the dataset\n",
    "records.first()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# `pyspark.RDD.groupBy`\n",
    "\n",
    "Using the groupBy operator to count the number of jobs per person..."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Group the records by the name of the person\n",
    "grouped_records = records.groupBy(lambda x: x[\"name\"])\n",
    "\n",
    "# Show the first group\n",
    "grouped_records.first()\n",
    "\n",
    "# Count the groups\n",
    "job_counts = grouped_records.map(\n",
    "  lambda x: {\n",
    "    \"name\": x[0],\n",
    "    \"job_count\": len(x[1])\n",
    "  }\n",
    ")\n",
    "\n",
    "job_counts.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Exercises\n",
    "\n",
    "1. Use `pyspark.RDD.groupBy` to group executives by job title, then prepare records with the job title and the count of the number of executives with that job."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Map vs FlatMap\n",
    "\n",
    "We need to understand the difference between the map and flatmap operators. Map groups items per-record, while flatMap creates a single large group of items."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Compute a relation of words by line\n",
    "words_by_line = csv_lines\\\n",
    "  .map(lambda line: line.split(\",\"))\n",
    "\n",
    "words_by_line.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Compute a relation of words\n",
    "flattened_words = csv_lines\\\n",
    "  .map(lambda line: line.split(\",\"))\\\n",
    "  .flatMap(lambda x: x)\n",
    "flattened_words.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "lengths = flattened_words.map(lambda x: len(x))\n",
    "lengths.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "lengths.sum() / lengths.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Creating Rows\n",
    "\n",
    "We can create `pyspark.sql.Rows` out of python objects so you we create `pyspark.sql.DataFrames`. This is desirable because once we have `DataFrames` we can run Spark SQL on our data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import Row\n",
    "\n",
    "# Convert the CSV into a pyspark.sql.Row\n",
    "def csv_to_row(line):\n",
    "  parts = line.split(\",\")\n",
    "  row = Row(\n",
    "    name=parts[0],\n",
    "    company=parts[1],\n",
    "    title=parts[2]\n",
    "  )\n",
    "  return row\n",
    "\n",
    "# Apply the function to get rows in an RDD\n",
    "rows = csv_lines.map(csv_to_row)\n",
    "rows.first()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Exercises\n",
    "\n",
    "1. First count the number of job titles for each executive, then create a `pyspark.sql.Row` for this result."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "records = csv_lines.map(lambda line: line.split(','))\n",
    "records.collect()\n",
    "\n",
    "groups = records.groupBy(lambda x: x[0])\n",
    "counts = groups.map(lambda x: (x[0], len(x[1])))\n",
    "new_rows = counts.map(lambda x: Row(name=x[0], count=x[1]))\n",
    "new_rows.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "rows.toDF().select(\"name\",\"count\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Creating `DataFrames` from `RDDs`\n",
    "\n",
    "Using the `RDD.toDF()` method to create a dataframe, registering the `DataFrame` as a temporary table with Spark SQL, and counting the jobs per person using Spark SQL."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Convert to a pyspark.sql.DataFrame\n",
    "rows_df = rows.toDF()\n",
    "\n",
    "rows_df.show()\n",
    "\n",
    "# Register the DataFrame for Spark SQL\n",
    "rows_df.registerTempTable(\"executives\")\n",
    "\n",
    "# Generate a new DataFrame with SQL using the SparkSession\n",
    "job_counts = spark.sql(\"\"\"\n",
    "SELECT\n",
    "  name,\n",
    "  COUNT(*) AS total\n",
    "  FROM executives\n",
    "  GROUP BY name\n",
    "\"\"\")\n",
    "job_counts.show()\n",
    "\n",
    "# Go back to an RDD\n",
    "job_counts.rdd.map(lambda x: x.asDict()).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## `SparkContext.parallelize()`\n",
    "\n",
    "The oppotiste of `pyspark.RDD.collect()` is `SparkContext.parallelize()`. Whereas `collect` pulls data from Spark's memory into local RAM, `parallelize` sends data from local memory to Spark's memory. \n",
    "\n",
    "You can access it like this: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "my_rdd = sc.parallelize([1,2,3,4,5])\n",
    "my_rdd.first()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Exercises\n",
    "\n",
    "1. Create your own RDD of `dict` elements with named fields using sc.parallelize. Make it at least 5 records long.\n",
    "2. Convert this RDD of `dicts` into an RDD of `pyspark.sql.Row` elements.\n",
    "3. Convert this RDD of `pyspark.sql.Rows` into a `pyspark.sql.DataFrame`.\n",
    "4. Run a SQL `GROUP BY`/`COUNT` on your new `DataFrame`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "my_data = [\n",
    "    {\"name\": \"Russell Jurney\", \"interest\": \"Ancient Greece\"},\n",
    "    {\"name\": \"Chris Jurney\", \"interest\": \"Virtual Reality\"},\n",
    "    {\"name\": \"Bill Jurney\", \"interest\": \"Sports\"},\n",
    "    {\"name\": \"Ruth Jurney\", \"interest\": \"Wildlife\"},\n",
    "    {\"name\": \"Bob Smith\", \"interest\": \"Sports\"}\n",
    "]\n",
    "my_rdd = sc.parallelize(my_data)\n",
    "my_rows = my_rdd.map(lambda x: Row(name=x[\"name\"], interest=x[\"interest\"]))\n",
    "\n",
    "my_df = my_rows.toDF()\n",
    "my_df.show()\n",
    "\n",
    "my_df.registerTempTable(\"people\")\n",
    "spark.sql(\"\"\"SELECT interest, COUNT(*) as total FROM people GROUP BY interest\"\"\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Creating `RDDs` from `DataFrames`\n",
    "\n",
    "We can easily convert back from a `DataFrame` to an `RDD` using the `pyspark.sql.DataFrame.rdd()` method, along with `pyspark.sql.Row.asDict()` if we desire a Python `dict` of our records."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "job_counts.rdd.map(lambda x: x.asDict()).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Exercises\n",
    "\n",
    "1. Using the data from item 4 in the exercise above, convert the data back to its original form, a local collection of `dict` elements."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "my_df.rdd.map(lambda x: x.asDict()).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Loading and Inspecting Parquet Files\n",
    "\n",
    "Using the `SparkSession` to load files as `DataFrames` and inspecting their contents..."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Load the parquet file containing flight delay records\n",
    "on_time_dataframe = spark.read.parquet('../data/on_time_performance.parquet')\n",
    "\n",
    "# Register the data for Spark SQL\n",
    "on_time_dataframe.registerTempTable(\"on_time_performance\")\n",
    "\n",
    "# Check out the columns\n",
    "on_time_dataframe.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Trim the fields and keep the result\n",
    "trimmed_on_time = on_time_dataframe\\\n",
    "  .select(\n",
    "    \"FlightDate\",\n",
    "    \"TailNum\",\n",
    "    \"Origin\",\n",
    "    \"Dest\",\n",
    "    \"Carrier\",\n",
    "    \"DepDelay\",\n",
    "    \"ArrDelay\"\n",
    "  )\n",
    "\n",
    "# Sample 0.01% of the data and show\n",
    "trimmed_on_time.sample(False, 0.0001).show(10)\n",
    "\n",
    "sampled_ten_percent = trimmed_on_time.sample(False, 0.1)\n",
    "sampled_ten_percent.show(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## `DataFrame` Workflow: Calculating Speed in Dataflow and SQL\n",
    "\n",
    "We can go back and forth between dataflow programming and SQL programming using `pyspark.sql.DataFrames`. This enables us to get the best of both worlds from these two APIs. For example, if we want to group records and get a total count for each group... a SQL `SELECT/GROUP BY/COUNT` is the most direct way to do it. On the other hand, if we want to filter data, a dataflow API call like `DataFrame.filter()` is the cleanest way. This comes down to personal preference for the user. In time you will develop your own style of working.\n",
    "\n",
    "### Dataflow Programming\n",
    "\n",
    "If we were to look at the `AirTime` along with the `Distance`, we could get a good idea of how fast the airplanes were going. Pretty cool! Lets do this using Dataflows first.\n",
    "\n",
    "#### Trimming Our Data\n",
    "\n",
    "First lets select just the two columns of interest: `AirTime` and `Distance`. We can always go back and select more columns if we want to extend our analysis, but trimming uneeded fields optimizes performance right away."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "fd = on_time_dataframe.select(\"AirTime\", \"Distance\")\n",
    "fd.show(6)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### From Minutes to Hours\n",
    "\n",
    "Now lets convert our `AirTime` from minutes to hours by dividing by 60."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "hourly_fd = fd.select((fd.AirTime / 60).alias('Hours'), \"Distance\")\n",
    "hourly_fd.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Raw Calculation\n",
    "\n",
    "Now lets calculate miles per hour!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "miles_per_hour = hourly_fd.select(\n",
    "    (hourly_fd.Distance / hourly_fd.Hours).alias('Mph')\n",
    ")\n",
    "miles_per_hour.show(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Investigating `nulls`\n",
    "\n",
    "Looks like we have some errors in some records in our calculation because of missing fields? Lets bring back in the `Distance` and `AirTime` fields to see where the problem is coming from."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "fd.select(\n",
    "    \"AirTime\", \n",
    "    (fd.AirTime / 60).alias('Hours'), \n",
    "    \"Distance\"\n",
    ").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Filtering `nulls`\n",
    "\n",
    "Now that we know some records are missing `AirTimes`, we can filter those records using `pyspark.sql.DataFrame.filter()`. Starting from the beginning, lets recalculate our values."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "fd = on_time_dataframe.select(\"AirTime\", \"Distance\")\n",
    "filled_fd = fd.filter(fd.AirTime.isNotNull())\n",
    "hourly_fd = filled_fd.select(\n",
    "    \"AirTime\", \n",
    "    (filled_fd.AirTime / 60).alias('Hours'), \n",
    "    \"Distance\"\n",
    ")\n",
    "mph = hourly_fd.select((hourly_fd.Distance / hourly_fd.Hours).alias('Mph'))\n",
    "mph.show(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Averaging Speed\n",
    "\n",
    "How fast does the fleet travel overall? Lets compute the average speed for the entire fleet."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import avg\n",
    "\n",
    "mph.select(\n",
    "    pyspark.sql.functions.avg(mph.Mph)\n",
    ").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "It looks like the average speed of the fleet is 408 mph. Note how along the way we chekced the data for sanity, which led to confidence in our answer. SQL by contast can hide the internals of a query, which might have skewed our average significantly! \n",
    "\n",
    "### SQL-Based Speed Calculation\n",
    "\n",
    "Now lets work the same thing out in SQL. Starting from the top:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "on_time_dataframe.registerTempTable(\"on_time_performance\")\n",
    "\n",
    "mph = spark.sql(\"\"\"\n",
    "SELECT ( Distance / ( AirTime/60 ) ) AS Mph \n",
    "FROM on_time_performance \n",
    "WHERE AirTime IS NOT NULL\n",
    "ORDER BY AirTime\n",
    "\"\"\")\n",
    "mph.show(10)\n",
    "\n",
    "mph.registerTempTable(\"mph\")\n",
    "spark.sql(\"SELECT AVG(Mph) from mph\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Evaluating SQL\n",
    "\n",
    "The SQL based solution seems to be better in this case, because we can simply express our calculation all in one place. When complexity grows however, it is best to break a single query into multiple stages where you use SQL or Dataflow programming to massage the data into shape."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Calculating Histograms\n",
    "\n",
    "Having computed the speed in miles per hour and the overall average speed of passenger jets in the US, lets dig deeper by using the `RDD` API's histogram method to calculate histograms buckets and values, which will then use to visualize data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Compute a histogram of departure delays\n",
    "mph\\\n",
    "  .select(\"Mph\")\\\n",
    "  .rdd\\\n",
    "  .flatMap(lambda x: x)\\\n",
    "  .histogram(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Visualizing Histograms\n",
    "\n",
    "The problem with the output above is that it is hard to interpret. For better understanding, we need a visualization. We can use `matplotlib` inline in a Jupyter Notebook to visualize this distribution and see what the tendency of speed of airplanes is around the mean of 408 mph."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%matplotlib inline\n",
    "\n",
    "import numpy as np\n",
    "import matplotlib.mlab as mlab\n",
    "import matplotlib.pyplot as plt\n",
    "\n",
    "# Function to plot a histogram using pyplot\n",
    "def create_hist(rdd_histogram_data):\n",
    "  \"\"\"Given an RDD.histogram, plot a pyplot histogram\"\"\"\n",
    "  heights = np.array(rdd_histogram_data[1])\n",
    "  full_bins = rdd_histogram_data[0]\n",
    "  mid_point_bins = full_bins[:-1]\n",
    "  widths = [abs(i - j) for i, j in zip(full_bins[:-1], full_bins[1:])]\n",
    "  bar = plt.bar(mid_point_bins, heights, width=widths, color='b')\n",
    "  return bar\n",
    "\n",
    "# Compute a histogram of departure delays\n",
    "departure_delay_histogram = mph\\\n",
    "  .select(\"Mph\")\\\n",
    "  .rdd\\\n",
    "  .flatMap(lambda x: x)\\\n",
    "  .histogram(10)\n",
    "\n",
    "create_hist(departure_delay_histogram)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Iterating on a Histogram\n",
    "\n",
    "That looks interesting, but the bars seem to fat the really see what is going on. Lets double the number of buckets from 10 to 20. We can reuse the `create_hist()` method to do so."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Compute a histogram of departure delays\n",
    "departure_delay_histogram = mph\\\n",
    "  .select(\"Mph\")\\\n",
    "  .rdd\\\n",
    "  .flatMap(lambda x: x)\\\n",
    "  .histogram(20)\n",
    "\n",
    "create_hist(departure_delay_histogram)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Speed Summary\n",
    "\n",
    "You've now seen how to calculate different values in both SQL and Dataflow style, how to switch between the two methods, how to switch between the `pyspark.RDD` and `pyspark.sql.DataFrame` APIs and you're starting to build a proficiency in PySpark!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Counting Airplanes in the US Fleet\n",
    "\n",
    "Lets convert our `on_time_dataframe` (a `DataFrame`) into an `RDD` to calculate the total number of airplanes in the US fleet."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Dump the unneeded fields\n",
    "tail_numbers = on_time_dataframe.rdd.map(lambda x: x.TailNum)\n",
    "tail_numbers = tail_numbers.filter(lambda x: x != '' and x is not None)\n",
    "\n",
    "# distinct() gets us unique tail numbers\n",
    "unique_tail_numbers = tail_numbers.distinct()\n",
    "\n",
    "# now we need a count() of unique tail numbers\n",
    "airplane_count = unique_tail_numbers.count()\n",
    "print(\"Total airplanes: {}\".format(airplane_count))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Exercise 1: Characterizing Airports\n",
    "\n",
    "Using the techniques we demonstated above, calculate any 3 out of 4 of the following things using both the SQL and the Dataflow methods for each one. That is: implement each calculation twice - once in SQL and once using Dataflows. Try to use both the `RDD` and `DataFrame` APIs as you work.\n",
    "\n",
    "1. How many airports are there in the united states?\n",
    "2. What is the average flight time for flights arriving in San Francisco (SFO)? What does the distribution of this value look like? Plot a histogram using the `create_hist` method shown above.\n",
    "3. Which American airport has the fastest out-bound speeds? What does the distribution of the flight speeds at this one airport look like? Plot a histogram using the `create_hist` method shown above.\n",
    "4. What were the worst travel dates in terms of overall delayed flights in the US in 2015?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#on_time_dataframe.select(\"Origin\").distinct().show(1)\n",
    "#spark.sql(\"SELECT DISTINCT(Origin) FROM on_time_performance\").show(1)\n",
    "\n",
    "#on_time_dataframe.filter(on_time_dataframe.Dest == \"SFO\").groupBy(\"Origin\").agg({\"AirTime\": \"avg\"}).show(5)\n",
    "\n",
    "origin_hour_dist = on_time_dataframe.filter(\n",
    "    on_time_dataframe.AirTime.isNotNull()\n",
    ").select(\n",
    "    \"Origin\", \n",
    "    (on_time_dataframe.AirTime/60).alias(\"Hours\"), \n",
    "    \"Distance\"\n",
    ")\n",
    "mph_origins = origin_hour_dist.select(\n",
    "    \"Origin\", \n",
    "    (origin_hour_dist.Distance / origin_hour_dist.Hours).alias(\"Mph\")\n",
    ")\n",
    "mph_origins.registerTempTable(\"mph_origins\")\n",
    "\n",
    "avg_speeds = mph_origins.groupBy(\"Origin\").agg({\"Mph\": \"avg\"}).alias(\"Mph\")\n",
    "avg_speeds.show()\n",
    "\n",
    "#avg_speeds.registerTempTable(\"avg_speeds\")\n",
    "#spark.sql(\"SELECT Origin, avg(Mph) FROM avg_speeds ORDER BY avg(Mph) DESC LIMIT 1\").show()\n",
    "\n",
    "# origin_speeds = mph_origins.rdd.map(lambda x: x.asDict()).collect()\n",
    "# len(origin_speeds)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "on_time_dataframe.columns"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Calculating with `DataFrame.groupBy`\n",
    "\n",
    "We can use Spark SQL to calculate things using `DataFrames`, but we can also group data and calculate as we did with `RDDs`. For a full list of methods you can apply to grouped `DataFrames`, see [the documentation for pyspark.sql.GroupedData](http://spark.apache.org/docs/latest/api/python/pyspark.sql.html#pyspark.sql.GroupedData). Below we will demonstrate some of these methods."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Calculate average of every numeric field\n",
    "on_time_dataframe.groupBy(\"Origin\").avg().show(1)\n",
    "\n",
    "# Calculate verage AirTime per origin city\n",
    "on_time_dataframe.groupBy(\"Origin\").agg({\"AirTime\": \"mean\"}).show(1)\n",
    "\n",
    "# Get the count of flights from each origin\n",
    "on_time_dataframe.groupBy(\"Origin\").count().show(1)\n",
    "\n",
    "# Get the maximum airtime for flights leaving each city\n",
    "on_time_dataframe.groupby(\"Origin\").agg({\"AirTime\": \"max\"}).show(1)\n",
    "\n",
    "# Get the maximum of all numeric columns for flights leaving each city\n",
    "on_time_dataframe.groupBy(\"Origin\").max().show(1)\n",
    "\n",
    "# Get the shortest flight for each origin airport\n",
    "on_time_dataframe.groupBy(\"Origin\").agg({\"AirTime\": \"min\"}).show(1)\n",
    "\n",
    "# Total minutes flown from each airport\n",
    "on_time_dataframe.groupBy(\"Origin\").agg({\"AirTime\": \"sum\"}).show(1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Pivoting DataFrames\n",
    "\n",
    "One useful function of `DataFrames` is `pivot`. Pivot lets you compute pivot tables from data. Lets use `pivot` to calculate the average flight times between Atlanta `ATL` and other airports."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "on_time_dataframe\\\n",
    "    .filter(\"Origin == 'ATL'\")\\\n",
    "    .groupBy(\"Origin\")\\\n",
    "    .pivot(\"Dest\")\\\n",
    "    .avg(\"AirTime\")\\\n",
    "    .rdd\\\n",
    "    .map(lambda x: x.asDict())\\\n",
    "    .collect()[0]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Plotting Scatterplots\n",
    "\n",
    "Another type of visualization that is of interest to data scientists is the scatterplot. A scatterplot enables us to compare the trend of one value plotted against the other. For example, we could calculate the relationship between `Origin` and `Dest` `Distance` and the `Mph` speed figure we calculated earlier. Are longer flights generally faster, or not?\n",
    "\n",
    "To prepare a scatterplot, we need to use `matplotlib` again, so we'll need to look at what its scatterplot API expects. The `matplotlib.pyplot.scatter` API takes two independant lists of values for the variables x and y, so we must compute them for `Distance` and `Mph`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "mph = spark.sql(\"\"\"\n",
    "SELECT \n",
    "    Distance, \n",
    "    ( Distance / ( AirTime/60 ) ) AS Mph \n",
    "FROM on_time_performance \n",
    "WHERE AirTime IS NOT NULL\n",
    "\"\"\")\n",
    "mph.show(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Collecting Data\n",
    "\n",
    "Note that we will have to convert our data from existing within our Spark cluster's memory to within our local computer's memory where `matplotlib` runs."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "distance = mph.select(\"Distance\").rdd.flatMap(lambda x: x)\n",
    "distance = distance.collect()\n",
    "distance[0:10]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "speed = mph.select(\"Mph\").rdd.flatMap(lambda x: x)\n",
    "speed = speed.collect()\n",
    "speed[0:10]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Sampling Data\n",
    "\n",
    "When I tried to plot this data, it took a very long time to draw. This is because... well, how many unique values are there for each variable? Lets see."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(\"Total distances: {:,}\".format(len(distance)))\n",
    "print(\"Total speeds: {:,}\".format(len(speed)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "It is hard to plot 5.7 million dots on a scatterplot that will fit on a computer screen. So lets sample our data. We can use PySpark `DataFrame's` sample method. Lets take a 0.1% random sample without replacement, which will leave us with 5,687 or so data points - something we can more easily manage."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sample = mph.sample(False, 0.001)\n",
    "sample.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Note that we need to sample __once__ and then split the datasets out - otherwise the data for a single observation will be scrambled across variables. We don't want that! All our scatterplots would show no relationships at all."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "speed = sample.select(\"Mph\").rdd.flatMap(lambda x: x).collect()\n",
    "distance = sample.select(\"Distance\").rdd.flatMap(lambda x: x).collect()\n",
    "print(\"{:,} x {:,} records!\".format(\n",
    "    len(speed),\n",
    "    len(distance)\n",
    "))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Fun with `matplotlib.pyplot.scatter`\n",
    "\n",
    "Now we feed the `scatter` API `distance` as `x` and `speed` as `y`, giving it a title and x and y axes. Note that we also specify a size in inches via the `figure.figsize` `rcParam`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%matplotlib inline\n",
    "\n",
    "import numpy as np\n",
    "import matplotlib.pyplot as plt\n",
    "\n",
    "plt.rcParams[\"figure.figsize\"] = (18,12)\n",
    "\n",
    "plt.scatter(\n",
    "    distance, \n",
    "    speed, \n",
    "    alpha=0.5\n",
    ")\n",
    "plt.title(\"Distance x Speed\")\n",
    "plt.xlabel(\"Distance\")\n",
    "plt.ylabel(\"Speed\")\n",
    "plt.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Interpreting Our Scatterplot\n",
    "\n",
    "We can see pretty clearly that as distance increases, average speed across that distance increases rapidly and then levels off as the distance increases.\n",
    "\n",
    "## Exercises\n",
    "\n",
    "1. Query the `on_time_dataframe` to focus on two numeric fields. \n",
    "2. Plot a histogram of one of these fields\n",
    "3. Plot a scatterplot of both of these fields\n",
    "\n",
    "# Predicting Speed Given Distance\n",
    "\n",
    "It is often the case that once we characterize a distribution, we want to create a function to predict one variable given the other. Lets take this example further by fitting a polynomial regression to describe our data. We use `sklearn.pipeline.Pipeline` to chain a `sklearn.preprocessing.PolynomialFeatures` to a `sklearn.linear_model.LinearRegression`. Other than that, we simply define x and y, and fit a model to those values. Then we finally compute a cross value score, to see the model's performance. We'll see this pattern again when we use large data tools in Spark MLlib."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sklearn.model_selection import train_test_split\n",
    "from sklearn.preprocessing import PolynomialFeatures\n",
    "from sklearn.linear_model import LinearRegression\n",
    "from sklearn.pipeline import Pipeline\n",
    "from sklearn.model_selection import cross_val_score\n",
    "import numpy as np\n",
    "\n",
    "x = np.array(distance)\n",
    "y = np.array(speed)\n",
    "x_test = np.arange(0, 5000, 100)\n",
    "\n",
    "model = Pipeline([\n",
    "    ('poly', PolynomialFeatures(degree=3)),\n",
    "    ('linear', LinearRegression(fit_intercept=False))\n",
    "])\n",
    "model = model.fit(x[:, np.newaxis], y)\n",
    "model.named_steps['linear'].coef_\n",
    "y_out = model.predict(x_test.reshape(-1,1))\n",
    "\n",
    "cross_val_score(model, x.reshape(-1,1), y)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Visualizing Polynomial Fit\n",
    "\n",
    "Because we are running a polynomial regression, where we get to decide the degree of the polynomial. To help decide, lets plot a polynomial fit line to the data using `matplotlib`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%matplotlib inline\n",
    "\n",
    "import numpy as np\n",
    "import matplotlib.pyplot as plt\n",
    "\n",
    "plt.rcParams[\"figure.figsize\"] = (18,12)\n",
    "\n",
    "plt.scatter(\n",
    "    distance, \n",
    "    speed, \n",
    "    alpha=0.5\n",
    ")\n",
    "plt.plot(\n",
    "    x_test,\n",
    "    y_out, \n",
    "    color='orange', \n",
    "    linewidth=3\n",
    ")\n",
    "\n",
    "plt.title(\"Distance x Speed\")\n",
    "plt.xlabel(\"Distance\")\n",
    "plt.ylabel(\"Speed\")\n",
    "plt.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Joining Data in PySpark\n",
    "\n",
    "Next we're going to learn how to join between datasets using PySpark. We're going to pick up from an example we're going to work in chapter 6, and explore it more deeply. To begin with, we will prepare a list of `TailNum` (tail numbers) from the FAA flight records. These uniquely identify each airplane from each flight.\n",
    "\n",
    "### Unique Tail Numbers"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "tail_numbers = on_time_dataframe.select(\"TailNum\").distinct()\n",
    "tail_numbers.show(6)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### FAA Airplane Records\n",
    "\n",
    "We will trim the FAA records down to just the `TailNum`, `Model` and `Engine_Model`. Note that we go ahead and rename the `TailNum` field to `FAATailNum` using the `pyspark.sql.functions.alias()` method. This avoids having two fields referenced by the same name once we perform our joins."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "faa_tail_number_inquiry = spark.read.json('../data/faa_tail_number_inquiry.jsonl')\n",
    "airplane_records = faa_tail_number_inquiry.select(\n",
    "    faa_tail_number_inquiry.TailNum.alias(\"FAATailNum\"), \n",
    "    \"Model\",\n",
    "    \"Engine_Model\"\n",
    ")\n",
    "airplane_records.show(6)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Inner Joins\n",
    "\n",
    "You may be familiar with an inner join from SQL. An inner join joins two datasets based on the presence of a key from one dataset in the other. Records which don't have a key that appears in the other table do not appear in the final output."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# INNER JOIN\n",
    "print(\n",
    "    \"FAA tail numbers: {:,}\".format(\n",
    "        tail_numbers.count()\n",
    "    )\n",
    ")\n",
    "print(\n",
    "    \"Airplane records: {:,}\".format(\n",
    "        airplane_records.count()\n",
    "    )\n",
    ")\n",
    "\n",
    "inner_joined = tail_numbers.join(\n",
    "    airplane_records, \n",
    "    tail_numbers.TailNum == airplane_records.FAATailNum, \n",
    "    'inner'\n",
    ")\n",
    "\n",
    "print(\n",
    "    \"Joined records:   {:,}\".format(\n",
    "        inner_joined.count()\n",
    "    )\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Inner Join Results\n",
    "\n",
    "Note that there are as many records in the output as there were in the FAA Airplane dataset - indicating that there was a representative of every tail number from that dataset in the on-time performance flight records. Lets take a look at the records themselves."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "inner_joined.show(6)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Note how convenient it is that we renamed one of the keys `FAATailNum`. If we hadn't, we'd have two columns with the same name now and would have trouble referring to one or the other."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Left Outer Join\n",
    "\n",
    "Another type of join is the left outer join. It ensures that one record will remain in the output from the left side of the join no matter what. If a match on the join keys is found, the fields for the record on the right will be filled. If a match is not found, they will be empty. \n",
    "\n",
    "Lets look at how this works with our two datasets."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# INNER JOIN\n",
    "print(\n",
    "    \"FAA tail numbers: {:,}\".format(\n",
    "        tail_numbers.count()\n",
    "    )\n",
    ")\n",
    "print(\n",
    "    \"Airplane records: {:,}\".format(\n",
    "        airplane_records.count()\n",
    "    )\n",
    ")\n",
    "\n",
    "left_outer_joined = tail_numbers.join(\n",
    "    airplane_records, \n",
    "    tail_numbers.TailNum == airplane_records.FAATailNum, \n",
    "    'left_outer'\n",
    ")\n",
    "\n",
    "print(\n",
    "    \"Joined records:   {:,}\".format(\n",
    "        left_outer_joined.count()\n",
    "    )\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Left Outer Join Result\n",
    "\n",
    "Note that there were 4,898 records on the left side of our join and there are the same number on the output of our join. Lets take a look at what both matched and unmatched records look like:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "left_outer_joined.show(6)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Note that some records have fields filled out, and some don't.\n",
    "\n",
    "## Right Outer Join\n",
    "\n",
    "Another type of join is the right outer join. This works the opposite of a left outer join. In this case, the output will preserve a record for each and every record on the right side of the join. Use the `right_outer` join type to perform this kind of join.\n",
    "\n",
    "## Exercises\n",
    "\n",
    "1. Go back and perform a right outer join on the preceding two datasets. Is the `distinct()` call on the FAA on-time performance records still needed? Why or why not?\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Using `RDDs` and Map/Reduce to Prepare a Complex Record"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Filter down to the fields we need to identify and link to a flight\n",
    "flights = on_time_dataframe.rdd.map(\n",
    "    lambda x: (x.Carrier, x.FlightDate, x.FlightNum, x.Origin, x.Dest, x.TailNum)\n",
    ")\n",
    "\n",
    "# Group flights by tail number, sorted by date, then flight number, then origin/dest\n",
    "flights_per_airplane = flights\\\n",
    "  .map(lambda nameTuple: (nameTuple[5], [nameTuple[0:5]]))\\\n",
    "  .reduceByKey(lambda a, b: a + b)\\\n",
    "  .map(lambda tuple:\n",
    "      {\n",
    "        'TailNum': tuple[0], \n",
    "        'Flights': sorted(tuple[1], key=lambda x: (x[1], x[2], x[3], x[4]))\n",
    "      }\n",
    "    )\n",
    "flights_per_airplane.first()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Counting Late Flights"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "total_flights = on_time_dataframe.count()\n",
    "\n",
    "# Flights that were late leaving...\n",
    "late_departures = on_time_dataframe.filter(\n",
    "  on_time_dataframe.DepDelayMinutes > 0\n",
    ")\n",
    "total_late_departures = late_departures.count()\n",
    "print(total_late_departures)\n",
    "\n",
    "# Flights that were late arriving...\n",
    "late_arrivals = on_time_dataframe.filter(\n",
    "  on_time_dataframe.ArrDelayMinutes > 0\n",
    ")\n",
    "total_late_arrivals = late_arrivals.count()\n",
    "print(total_late_arrivals)\n",
    "\n",
    "# Get the percentage of flights that are late, rounded to 1 decimal place\n",
    "pct_late = round((total_late_arrivals / (total_flights * 1.0)) * 100, 1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Counting Flights with Hero Captains\n",
    "\n",
    "\"Hero Captains\" are those that depart late but make up time in the air and arrive on time or early."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Flights that left late but made up time to arrive on time...\n",
    "on_time_heros = on_time_dataframe.filter(\n",
    "  (on_time_dataframe.DepDelayMinutes > 0)\n",
    "  &\n",
    "  (on_time_dataframe.ArrDelayMinutes <= 0)\n",
    ")\n",
    "total_on_time_heros = on_time_heros.count()\n",
    "print(total_on_time_heros)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Printing Our Results"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(\"Total flights:   {:,}\".format(total_flights))\n",
    "print(\"Late departures: {:,}\".format(total_late_departures))\n",
    "print(\"Late arrivals:   {:,}\".format(total_late_arrivals))\n",
    "print(\"Recoveries:      {:,}\".format(total_on_time_heros))\n",
    "print(\"Percentage Late: {}%\".format(pct_late))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Computing the Average Lateness Per Flights"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Get the average minutes late departing and arriving\n",
    "spark.sql(\"\"\"\n",
    "SELECT\n",
    "  ROUND(AVG(DepDelay),1) AS AvgDepDelay,\n",
    "  ROUND(AVG(ArrDelay),1) AS AvgArrDelay\n",
    "FROM on_time_performance\n",
    "\"\"\"\n",
    ").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Inspecting Late Flights"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Why are flights late? Lets look at some delayed flights and the delay causes\n",
    "late_flights = spark.sql(\"\"\"\n",
    "SELECT\n",
    "  ArrDelayMinutes,\n",
    "  WeatherDelay,\n",
    "  CarrierDelay,\n",
    "  NASDelay,\n",
    "  SecurityDelay,\n",
    "  LateAircraftDelay\n",
    "FROM\n",
    "  on_time_performance\n",
    "WHERE\n",
    "  WeatherDelay IS NOT NULL\n",
    "  OR\n",
    "  CarrierDelay IS NOT NULL\n",
    "  OR\n",
    "  NASDelay IS NOT NULL\n",
    "  OR\n",
    "  SecurityDelay IS NOT NULL\n",
    "  OR\n",
    "  LateAircraftDelay IS NOT NULL\n",
    "ORDER BY\n",
    "  FlightDate\n",
    "\"\"\")\n",
    "late_flights.sample(False, 0.01).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Determining Why Flights Are Late"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Calculate the percentage contribution to delay for each source\n",
    "total_delays = spark.sql(\"\"\"\n",
    "SELECT\n",
    "  ROUND(SUM(WeatherDelay)/SUM(ArrDelayMinutes) * 100, 1) AS pct_weather_delay,\n",
    "  ROUND(SUM(CarrierDelay)/SUM(ArrDelayMinutes) * 100, 1) AS pct_carrier_delay,\n",
    "  ROUND(SUM(NASDelay)/SUM(ArrDelayMinutes) * 100, 1) AS pct_nas_delay,\n",
    "  ROUND(SUM(SecurityDelay)/SUM(ArrDelayMinutes) * 100, 1) AS pct_security_delay,\n",
    "  ROUND(SUM(LateAircraftDelay)/SUM(ArrDelayMinutes) * 100, 1) AS pct_late_aircraft_delay\n",
    "FROM on_time_performance\n",
    "\"\"\")\n",
    "total_delays.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Computing a Histogram of Weather Delayed Flights"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Eyeball the first to define our buckets\n",
    "weather_delay_histogram = on_time_dataframe\\\n",
    "  .select(\"WeatherDelay\")\\\n",
    "  .rdd\\\n",
    "  .flatMap(lambda x: x)\\\n",
    "  .histogram([1, 5, 10, 15, 30, 60, 120, 240, 480, 720, 24*60.0])\n",
    "print(weather_delay_histogram)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "create_hist(weather_delay_histogram)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Preparing a Histogram for Visualization by d3.js"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Transform the data into something easily consumed by d3\n",
    "def histogram_to_publishable(histogram):\n",
    "  record = {'key': 1, 'data': []}\n",
    "  for label, value in zip(histogram[0], histogram[1]):\n",
    "    record['data'].append(\n",
    "      {\n",
    "        'label': label,\n",
    "        'value': value\n",
    "      }\n",
    "    )\n",
    "  return record\n",
    "\n",
    "# Recompute the weather histogram with a filter for on-time flights\n",
    "weather_delay_histogram = on_time_dataframe\\\n",
    "  .filter(\n",
    "    (on_time_dataframe.WeatherDelay != None)\n",
    "    &\n",
    "    (on_time_dataframe.WeatherDelay > 0)\n",
    "  )\\\n",
    "  .select(\"WeatherDelay\")\\\n",
    "  .rdd\\\n",
    "  .flatMap(lambda x: x)\\\n",
    "  .histogram([0, 15, 30, 60, 120, 240, 480, 720, 24*60.0])\n",
    "print(weather_delay_histogram)\n",
    "\n",
    "record = histogram_to_publishable(weather_delay_histogram)\n",
    "record"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Building a Classifier Model to Predict Flight Delays\n",
    "\n",
    "## Loading Our Data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.types import StringType, IntegerType, FloatType, DoubleType, DateType, TimestampType\n",
    "from pyspark.sql.types import StructType, StructField\n",
    "from pyspark.sql.functions import udf\n",
    "\n",
    "schema = StructType([\n",
    "  StructField(\"ArrDelay\", DoubleType(), True),     # \"ArrDelay\":5.0\n",
    "  StructField(\"CRSArrTime\", TimestampType(), True),    # \"CRSArrTime\":\"2015-12-31T03:20:00.000-08:00\"\n",
    "  StructField(\"CRSDepTime\", TimestampType(), True),    # \"CRSDepTime\":\"2015-12-31T03:05:00.000-08:00\"\n",
    "  StructField(\"Carrier\", StringType(), True),     # \"Carrier\":\"WN\"\n",
    "  StructField(\"DayOfMonth\", IntegerType(), True), # \"DayOfMonth\":31\n",
    "  StructField(\"DayOfWeek\", IntegerType(), True),  # \"DayOfWeek\":4\n",
    "  StructField(\"DayOfYear\", IntegerType(), True),  # \"DayOfYear\":365\n",
    "  StructField(\"DepDelay\", DoubleType(), True),     # \"DepDelay\":14.0\n",
    "  StructField(\"Dest\", StringType(), True),        # \"Dest\":\"SAN\"\n",
    "  StructField(\"Distance\", DoubleType(), True),     # \"Distance\":368.0\n",
    "  StructField(\"FlightDate\", DateType(), True),    # \"FlightDate\":\"2015-12-30T16:00:00.000-08:00\"\n",
    "  StructField(\"FlightNum\", StringType(), True),   # \"FlightNum\":\"6109\"\n",
    "  StructField(\"Origin\", StringType(), True),      # \"Origin\":\"TUS\"\n",
    "])\n",
    "\n",
    "features = spark.read.json(\n",
    "  \"../data/simple_flight_delay_features.jsonl.bz2\",\n",
    "  schema=schema\n",
    ")\n",
    "features.first()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Check Data for Nulls"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#\n",
    "# Check for nulls in features before using Spark ML\n",
    "#\n",
    "null_counts = [(column, features.where(features[column].isNull()).count()) for column in features.columns]\n",
    "cols_with_nulls = filter(lambda x: x[1] > 0, null_counts)\n",
    "print(list(cols_with_nulls))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Add a Route Column\n",
    "\n",
    "Demonstrating the addition of a feature to our model..."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#\n",
    "# Add a Route variable to replace FlightNum\n",
    "#\n",
    "from pyspark.sql.functions import lit, concat\n",
    "\n",
    "features_with_route = features.withColumn(\n",
    "  'Route',\n",
    "  concat(\n",
    "    features.Origin,\n",
    "    lit('-'),\n",
    "    features.Dest\n",
    "  )\n",
    ")\n",
    "features_with_route.select(\"Origin\", \"Dest\", \"Route\").show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Bucketizing ArrDelay into ArrDelayBucket"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#\n",
    "# Use pysmark.ml.feature.Bucketizer to bucketize ArrDelay\n",
    "#\n",
    "from pyspark.ml.feature import Bucketizer\n",
    "\n",
    "splits = [-float(\"inf\"), -15.0, 0, 30.0, float(\"inf\")]\n",
    "bucketizer = Bucketizer(\n",
    "  splits=splits,\n",
    "  inputCol=\"ArrDelay\",\n",
    "  outputCol=\"ArrDelayBucket\"\n",
    ")\n",
    "ml_bucketized_features = bucketizer.transform(features_with_route)\n",
    "\n",
    "# Check the buckets out\n",
    "ml_bucketized_features.select(\"ArrDelay\", \"ArrDelayBucket\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Indexing Our String Fields into Numeric Fields"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#\n",
    "# Extract features tools in with pyspark.ml.feature\n",
    "#\n",
    "from pyspark.ml.feature import StringIndexer, VectorAssembler\n",
    "\n",
    "# Turn category fields into categoric feature vectors, then drop intermediate fields\n",
    "for column in [\"Carrier\", \"DayOfMonth\", \"DayOfWeek\", \"DayOfYear\",\n",
    "               \"Origin\", \"Dest\", \"Route\"]:\n",
    "  string_indexer = StringIndexer(\n",
    "    inputCol=column,\n",
    "    outputCol=column + \"_index\"\n",
    "  )\n",
    "  ml_bucketized_features = string_indexer.fit(ml_bucketized_features)\\\n",
    "                                          .transform(ml_bucketized_features)\n",
    "\n",
    "# Check out the indexes\n",
    "ml_bucketized_features.show(6)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Combining Numeric Fields into a Single Vector"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Handle continuous, numeric fields by combining them into one feature vector\n",
    "numeric_columns = [\"DepDelay\", \"Distance\"]\n",
    "index_columns = [\"Carrier_index\", \"DayOfMonth_index\",\n",
    "                   \"DayOfWeek_index\", \"DayOfYear_index\", \"Origin_index\",\n",
    "                   \"Origin_index\", \"Dest_index\", \"Route_index\"]\n",
    "vector_assembler = VectorAssembler(\n",
    "  inputCols=numeric_columns + index_columns,\n",
    "  outputCol=\"Features_vec\"\n",
    ")\n",
    "final_vectorized_features = vector_assembler.transform(ml_bucketized_features)\n",
    "\n",
    "# Drop the index columns\n",
    "for column in index_columns:\n",
    "  final_vectorized_features = final_vectorized_features.drop(column)\n",
    "\n",
    "# Check out the features\n",
    "final_vectorized_features.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Training Our Model in an Experimental Setup"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#\n",
    "# Cross validate, train and evaluate classifier\n",
    "#\n",
    "\n",
    "# Test/train split\n",
    "training_data, test_data = final_vectorized_features.randomSplit([0.7, 0.3])\n",
    "\n",
    "# Instantiate and fit random forest classifier\n",
    "from pyspark.ml.classification import RandomForestClassifier\n",
    "rfc = RandomForestClassifier(\n",
    "  featuresCol=\"Features_vec\",\n",
    "  labelCol=\"ArrDelayBucket\",\n",
    "  maxBins=4657,\n",
    "  maxMemoryInMB=1024\n",
    ")\n",
    "model = rfc.fit(training_data)\n",
    "\n",
    "# Evaluate model using test data\n",
    "predictions = model.transform(test_data)\n",
    "\n",
    "from pyspark.ml.evaluation import MulticlassClassificationEvaluator\n",
    "evaluator = MulticlassClassificationEvaluator(labelCol=\"ArrDelayBucket\", metricName=\"accuracy\")\n",
    "accuracy = evaluator.evaluate(predictions)\n",
    "print(\"Accuracy = {}\".format(accuracy))\n",
    "\n",
    "# Check a sample\n",
    "predictions.sample(False, 0.001, 18).orderBy(\"CRSDepTime\").show(6)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}
